package cn.bitkit.component.task;

import cn.bitkit.tools.common.ThreadUtil;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;

@Slf4j
public abstract class BaseTaskQueueService<T, R extends BaseTaskData> extends BaseTaskService {

    private final Queue<T> dataList = new ConcurrentLinkedDeque<>();
    private final ThreadFactory factory = Thread.ofVirtual().factory();
    volatile boolean readRunning = true;

    @Override
    public void doTask() {
        Thread readThread = factory.newThread(() -> {
            try {
                asyncReadData();
                ThreadUtil.sleep(1000);
            } catch (Exception e) {
                log.error(e.getMessage(), e);
            } finally {
                readRunning = false;
            }
        });
        readThread.start();
        dealData();
    }


    public void asyncReadData(){
        while (isRunning()){
            while(isRunning() && dataList.size() > taskData.getThreads() * 10){
                ThreadUtil.sleep(500);
            }
            List<T> list = queryData();
            dataList.addAll(list);
        }
    }

    @SneakyThrows
    public void dealData(){
        Semaphore semaphore = new Semaphore(taskData.getThreads(), true);
        while (isRunning()){
            while (isRunning() && dataList.isEmpty() && readRunning){
                ThreadUtil.sleep(100);
            }
            T data = dataList.poll();
            if(data == null){
                break;
            }
            semaphore.acquire();
            if (!isRunning()){
                break;
            }
            Thread virtualThread = factory.newThread(() -> {
                try {
                    processData(data);
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }finally {
                    semaphore.release();
                }
            });
            virtualThread.start();
        }
        //等待全部完成
        while (semaphore.availablePermits() != taskData.getThreads()){
            ThreadUtil.sleep(1000);
        }
    }

    /**
     * 查询数据
     * @return 数据
     */
    public abstract List<T> queryData();

    /**
     * 处理数据
     * @param data 数据
     */
    public abstract void processData(T data);

}
